package com.atguigu.day04;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.time.Duration;

public class Example5 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env.addSource(new SourceFunction<Tuple2<String,Long>>() {
            @Override
            public void run(SourceContext<Tuple2<String, Long>> ctx) throws Exception {
                ctx.collect(Tuple2.of("a", 1000L));
                Thread.sleep(1000L);
                ctx.collect(Tuple2.of("a", 2000L));
                Thread.sleep(1000L);
                ctx.collect(Tuple2.of("a", 3000L));
                Thread.sleep(1000L);
                ctx.collect(Tuple2.of("a", 1000L));
            }

            @Override
            public void cancel() {

            }
        })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Tuple2<String,Long>>forBoundedOutOfOrderness(Duration.ofMillis(1000))
                        .withTimestampAssigner((SerializableTimestampAssigner<Tuple2<String, Long>>) (stringLongTuple2, l) -> stringLongTuple2.f1)
                )
                .keyBy(x->x.f0)
                .process(new KeyedProcessFunction<String, Tuple2<String, Long>, String>() {
                    @Override
                    public void processElement(Tuple2<String, Long> value, Context ctx, Collector<String> out) throws Exception {
                        if (ctx.timestamp()<ctx.timerService().currentWatermark()){
                            out.collect("数据迟到了"+value+"时间是："+ctx.timestamp());
                        }else {
                            out.collect("正常数据："+value+"时间是:"+ctx.timestamp());
                        }
                    }
                })
                .print();

        env.execute();
    }
}
